using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class RabbitMQBatchClient : IDisposable
{
    private ConnectionFactory _factory;
    private IConnection _connection;
    private IModel _channel;
    private EventingBasicConsumer _consumer;
    private ConcurrentQueue<BasicDeliverEventArgs> _messageQueue = new ConcurrentQueue<BasicDeliverEventArgs>();
    private bool _isConsuming;
    private ulong _lastDeliveryTag;

    // 优化后的连接配置
    public void Connect(string hostName = "localhost")
    {
        _factory = new ConnectionFactory
        {
            HostName = hostName,
            SocketWriteTimeout = TimeSpan.FromMilliseconds(50),
            RequestedHeartbeat = TimeSpan.Zero,
            AutomaticRecoveryEnabled = false
        };
        _connection = _factory.CreateConnection();
        _channel = _connection.CreateModel();
    }

    // 高性能发布方法
    public void HighPerfPublish(string exchange, string routingKey,
        IEnumerable<byte[]> messages, int batchSize = 100_000)
    {
        var props = _channel.CreateBasicProperties();
        var batch = new byte[batchSize][];

        int index = 0;
        foreach (var body in messages)
        {
            batch[index++] = body;
            if (index >= batchSize)
            {
                UnsafePublishBatch(exchange, routingKey, batch, props, index);
                index = 0;
            }
        }

        if (index > 0)
        {
            UnsafePublishBatch(exchange, routingKey, batch, props, index);
        }
    }

    private unsafe void UnsafePublishBatch(string exchange, string routingKey,
       byte[][] batch, IBasicProperties props, int count)
    {
        // 使用指针数组作为中间载体
        var pointers = new byte*[count];

        // 第一阶段：固定所有内存块
        for (int i = 0; i < count; i++)
        {
            fixed (byte* ptr = batch[i])
            {
                pointers[i] = ptr;
            }
        }

        // 第二阶段：批量发布
        fixed (byte** ptrArray = &pointers[0])
        {
            for (int i = 0; i < count; i++)
            {
                var memory = new ReadOnlyMemory<byte>(batch[i]); // 使用原始数组
                _channel.BasicPublish(exchange, routingKey, props, memory);
            }
        }
    }

    // 批量消费实现
    public void StartBatchConsume(string queueName, int batchSize = 1,
        bool autoAck = false, Action<List<BasicDeliverEventArgs>> onBatchReceived = null)
    {
        _isConsuming = true;
        _consumer = new EventingBasicConsumer(_channel);

        _channel.BasicQos(0, (ushort)batchSize, false);

        _consumer.Received += (model, ea) =>
        {
            _messageQueue.Enqueue(ea);
            if (_messageQueue.Count >= batchSize)
            {
                ProcessBatch(autoAck, onBatchReceived);
            }
        };

        Task.Run(() =>
        {
            while (_isConsuming)
            {
                if (_messageQueue.Count > 0)
                {
                    ProcessBatch(autoAck, onBatchReceived);
                }
                Thread.Sleep(1);
            }
        });

        _channel.BasicConsume(queue: queueName, autoAck: autoAck, consumer: _consumer);
    }

    private void ProcessBatch(bool autoAck, Action<List<BasicDeliverEventArgs>> callback)
    {
        var batch = new List<BasicDeliverEventArgs>();
        while (batch.Count < 50000 && _messageQueue.TryDequeue(out var item))
        {
            batch.Add(item);
            _lastDeliveryTag = item.DeliveryTag;
        }

        callback?.Invoke(batch);

        if (!autoAck && batch.Count > 0)
        {
            _channel.BasicAck(_lastDeliveryTag, multiple: true);
        }
    }

    public void StopBatchConsume()
    {
        _isConsuming = false;
        ProcessBatch(false, null);
    }

    public void Dispose()
    {
        StopBatchConsume();
        _channel?.Close();
        _connection?.Close();
        GC.SuppressFinalize(this);
    }

    // 启用确认模式
    public void EnableConfirmMode()
    {
        _channel.ConfirmSelect();
    }
}
